package com.gator.rabbitmq.pablo_rabbitmq.dead_queue;


import com.gator.rabbitmq.pablo_rabbitmq.utils.ConnectionUtils;
import com.rabbitmq.client.CancelCallback;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;

import java.util.HashMap;
import java.util.Map;

/**
 * @Author PABLO
 * @Date 2022/6/5 21:35
 * @Desc 本消费者中有两个交换机 正常路由交换机和处理死信队列交换机
 */
public class Consumer {
    public static final String EXCHANGE_NAME = "pablo-exchange-normal";
    public static final String DEAD_EXCHANGE_NAME = "pablo-exchange-dead";
    public static final String EXCHANGE_TYPE = "direct";
    public static final String QUEUE = "normal-queue";
    public static final String DEAD_QUEUE = "dead-queue";
    public static final String NORMAL_ROUTING_KEY = "normal-routing-key";
    public static final String DEAD_ROUTING_KEY = "dead-routing-key";

    public static void main(String[] args) {
        Channel channel = ConnectionUtils.getConnectionAndChannel();
        try {
            //声明交换机
            channel.exchangeDeclare(EXCHANGE_NAME, EXCHANGE_TYPE);
            //死信交换机
            channel.exchangeDeclare(DEAD_EXCHANGE_NAME, EXCHANGE_TYPE);
            //声明队列
            channel.queueDeclare(QUEUE, true, false, false, setUpDeadQueue());
            //声明死信队列
            channel.queueDeclare(DEAD_QUEUE, true, false, false, null);
            //队列和交换机的绑定
            channel.queueBind(QUEUE, EXCHANGE_NAME, NORMAL_ROUTING_KEY);
            //死信队列交换机绑定
            channel.queueBind(DEAD_QUEUE, DEAD_EXCHANGE_NAME, DEAD_ROUTING_KEY);

            //开启手动应答
            channel.confirmSelect();
            //正常消费
            DeliverCallback deliverCallback = (consumerTag, message) -> {
                String msg = new String(message.getBody());
                //拒绝指定消息
                if (msg.equals("info-6")) {
                    System.out.println("consumer1拒收消息-->" + new String(message.getBody()));
                    //拒收并且不放入普通业务队列
                    channel.basicReject(message.getEnvelope().getDeliveryTag(),false);
                } else {
                    System.out.println("consumer1" + new String(message.getBody()));
                    //正常应答，并且是单个应答
                    channel.basicAck(message.getEnvelope().getDeliveryTag(),false);
                }
            };
            //消费者取消消费
            CancelCallback cancelCallback = consumerTag -> {
                System.out.println("msg被中断...");
            };
            channel.basicConsume(QUEUE, false, deliverCallback, cancelCallback);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public static Map setUpDeadQueue() {
        //处理死信队列
        Map<String, Object> arguments = new HashMap<>();
        //将超时信息转发给死信交换机
        arguments.put("x-dead-letter-exchange", DEAD_EXCHANGE_NAME);
        //死信交换机对应的路由key
        arguments.put("x-dead-letter-routing-key", DEAD_ROUTING_KEY);
        //设置普通队列长度，如超出则进入死信队列
        arguments.put("x-max-length", 5);
        return arguments;
    }

}

/**
 * @Author PABLO
 * @Date 2022/6/5 21:49
 * @Desc 处理死信队列中的数据
 */
class Consumer2 {

    public static final String DEAD_QUEUE = "dead-queue";

    public static void main(String[] args) {
        Channel channel = ConnectionUtils.getConnectionAndChannel();
        try {
            //正常消费
            DeliverCallback deliverCallback = (consumerTag, message) -> {
                System.out.println("consumer2消费死信队列消息" + new String(message.getBody()));
            };
            //消费者取消消费
            CancelCallback cancelCallback = consumerTag -> {
                System.out.println("msg被中断...");
            };
            channel.basicConsume(DEAD_QUEUE, true, deliverCallback, cancelCallback);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

}
